package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.DimUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Set;

/*
通过Phoenix 向Hbase插入数据
 */
public class DimSink extends RichSinkFunction<JSONObject> {
    private Connection connection = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        this.connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    // 生成语句提交HBase

    @Override
    public void invoke(JSONObject jsonObject, Context context) throws Exception {
        String table = jsonObject.getString("sink_table");
        JSONObject dataJsonObj = jsonObject.getJSONObject("data");
        // TODO  BUG： insert update 和 delete都拥有同样的data格式：但是因为在配置表中没有关于delete的配置，因此这了可以不需要delete
        if(dataJsonObj != null && dataJsonObj.size() >0) {
            String upsetSql = genUpsertSql(table.toUpperCase(),jsonObject.getJSONObject("data"));
            try {
                System.out.println(upsetSql);
                PreparedStatement ps = connection.prepareStatement(upsetSql);
                ps.executeUpdate();
                connection.commit();
                ps.close();
            } catch (Exception e){
                e.printStackTrace();
                throw new RuntimeException("执行插入Hbase sql失败");
            }
        }

        //如果维度数据发生了变化，那么清空当前数据在redis中的缓存
        if(jsonObject.getString("type").equals("update") ||
                jsonObject.getString("type").equals("delete")){
            assert dataJsonObj != null;
            DimUtil.deleteCached(table,dataJsonObj.getString("id"));
        }
    }

    public String genUpsertSql(String tableName,JSONObject jsonObject){
        Set<String> fields = jsonObject.keySet();
        String upsetSql = "upsert into "+ GmallConfig.HBASE_SCHEMA +"." + tableName +"(" + StringUtils.join(fields,",") +")";
        String valueSql = " values ('"+ StringUtils.join(jsonObject.values() ,"','") + "')";
        return upsetSql + valueSql;
    }
}
